//! 集群Redis网络层实现
//! 
//! 在简化Tokio网络层的基础上，添加CLUSTER命令支持

use std::sync::atomic::AtomicU64;
use std::io;
use tokio::net::{TcpStream, TcpListener};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use std::sync::atomic::Ordering;
use async_trait::async_trait;
use crate::core::network::AsyncNetworkStream;
use crate::storage::GlobalStorageManager;
use crate::event::ConnectionId;
use crate::cluster::{ClusterProtocolParser, RedisClusterCommand, ClusterCoordinator};
use std::sync::Arc;
use tokio::sync::RwLock;

/// 集群Redis网络流适配器
#[derive(Debug)]
pub struct ClusterTokioStream {
    /// 内部的Tokio TcpStream
    stream: TcpStream,
    /// 远程地址
    remote_addr: String,
}

impl ClusterTokioStream {
    pub fn from_tokio_stream(tokio_stream: TcpStream) -> io::Result<Self> {
        let remote_addr = tokio_stream.peer_addr()
            .map(|addr| addr.to_string())
            .unwrap_or_else(|_| "unknown".to_string());
        
        Ok(Self {
            stream: tokio_stream,
            remote_addr,
        })
    }
}

#[async_trait]
impl AsyncNetworkStream for ClusterTokioStream {
    async fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        self.stream.read(buf).await
    }
    
    async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        self.stream.write(buf).await
    }
    
    async fn flush(&mut self) -> io::Result<()> {
        self.stream.flush().await
    }
    
    fn is_connected(&self) -> bool {
        self.stream.peer_addr().is_ok()
    }
    
    fn remote_addr(&self) -> Option<String> {
        Some(self.remote_addr.clone())
    }
    
    async fn shutdown(&mut self) -> io::Result<()> {
        self.stream.shutdown().await
    }
    
    fn network_type(&self) -> &'static str {
        "Cluster-Tokio-TCP"
    }
}

/// 集群Redis协议处理器
pub struct ClusterProtocolHandler {
    _connection_id: ConnectionId,
    cluster_coordinator: Option<Arc<RwLock<ClusterCoordinator>>>,
}

impl ClusterProtocolHandler {
    pub fn new(connection_id: ConnectionId) -> Self {
        Self {
            _connection_id: connection_id,
            cluster_coordinator: None,
        }
    }
    
    pub fn with_coordinator(mut self, coordinator: Arc<RwLock<ClusterCoordinator>>) -> Self {
        self.cluster_coordinator = Some(coordinator);
        self
    }
    
    /// 处理连接流
    pub async fn handle_connection_stream(
        &mut self,
        mut stream: Box<dyn AsyncNetworkStream + Send>,
    ) -> io::Result<()> {
        let mut buffer = vec![0u8; 4096];
        let mut cmd_buffer = String::new();
        
        loop {
            // 读取数据
            let bytes_read = stream.read(&mut buffer).await?;
            if bytes_read == 0 {
                break; // 连接关闭
            }
            
            // 将数据添加到命令缓冲区
            if let Ok(data) = std::str::from_utf8(&buffer[..bytes_read]) {
                cmd_buffer.push_str(data);
                
                // 处理完整的命令
                while let Some(pos) = self.find_complete_command(&cmd_buffer) {
                    let command_str = cmd_buffer[..pos].to_string();
                    cmd_buffer.drain(..pos);
                    
                    let response = self.process_command(&command_str).await;
                    stream.write(response.as_bytes()).await?;
                    stream.flush().await?;
                }
            }
        }
        
        Ok(())
    }
    
    /// 查找完整的命令
    fn find_complete_command(&self, buffer: &str) -> Option<usize> {
        // 简化的RESP协议解析
        if buffer.starts_with('*') {
            // 查找完整的数组命令
            let lines: Vec<&str> = buffer.split("\r\n").collect();
            if lines.len() < 2 {
                return None;
            }
            
            // 解析数组长度
            if let Ok(array_len) = lines[0][1..].parse::<usize>() {
                let mut pos = 0;
                let mut current_line = 1;
                
                for _ in 0..array_len {
                    if current_line >= lines.len() {
                        return None;
                    }
                    
                    // 跳过长度行和数据行
                    current_line += 2;
                    if current_line > lines.len() {
                        return None;
                    }
                }
                
                // 计算总字节数
                for i in 0..current_line {
                    pos += lines[i].len() + 2; // +2 for \r\n
                }
                
                return Some(pos);
            }
        }
        
        None
    }
    
    /// 处理命令
    async fn process_command(&self, command_str: &str) -> String {
        // 解析RESP命令
        let args = self.parse_resp_command(command_str);
        if args.is_empty() {
            return "-ERR empty command\r\n".to_string();
        }
        
        let cmd_name = args[0].to_uppercase();
        let cmd_args = &args[1..];
        
        // 检查是否是CLUSTER命令
        if cmd_name == "CLUSTER" {
            return self.handle_cluster_command(cmd_args).await;
        }
        
        // 处理其他Redis命令（使用简化实现）
        self.handle_redis_command(&cmd_name, cmd_args).await
    }
    
    /// 处理CLUSTER命令
    async fn handle_cluster_command(&self, args: &[String]) -> String {
        if args.is_empty() {
            return "-ERR CLUSTER command requires subcommand\r\n".to_string();
        }
        
        // 使用ClusterProtocolParser解析命令
        match ClusterProtocolParser::parse_cluster_command(args) {
            Ok(cluster_cmd) => {
                // 如果有协调器，使用它处理命令
                if let Some(coordinator) = &self.cluster_coordinator {
                    self.execute_cluster_command_with_coordinator(cluster_cmd, coordinator).await
                } else {
                    // 没有协调器时的默认响应
                    self.execute_cluster_command_standalone(cluster_cmd).await
                }
            }
            Err(e) => {
                format!("-ERR {}\r\n", e)
            }
        }
    }
    
    /// 使用协调器执行集群命令
    async fn execute_cluster_command_with_coordinator(
        &self,
        cluster_cmd: RedisClusterCommand,
        coordinator: &Arc<RwLock<ClusterCoordinator>>,
    ) -> String {
        let coord = coordinator.read().await;
        
        match cluster_cmd {
            RedisClusterCommand::Info => {
                // 获取集群信息
                match coord.get_cluster_info().await {
                    Ok(info) => {
                        format!(
                            "+cluster_state:{}\r\ncluster_slots_assigned:{}\r\ncluster_known_nodes:{}\r\ncluster_size:{}\r\n", 
                            info.cluster_state,
                            info.slots_assigned,
                            info.node_count,
                            info.node_count
                        )
                    }
                    Err(_) => "-ERR failed to get cluster info\r\n".to_string(),
                }
            }
            RedisClusterCommand::Nodes => {
                // 获取集群节点信息
                match coord.get_cluster_nodes().await {
                    Ok(nodes_info) => {
                        format!("${}\r\n{}\r\n", nodes_info.len(), nodes_info)
                    }
                    Err(_) => "-ERR failed to get cluster nodes\r\n".to_string(),
                }
            }
            RedisClusterCommand::Slots => {
                // 使用协调器获取真实的槽位分配信息
                self.get_cluster_slots_from_coordinator(coordinator).await
            }
            RedisClusterCommand::MyId => {
                let node_id = coord.get_local_node_id().await;
                format!("${}\r\n{}\r\n", node_id.len(), node_id)
            }
            RedisClusterCommand::KeySlot { key } => {
                // 计算键的槽位
                let slot = self.calculate_slot(&key);
                format!(":{}\r\n", slot)
            }
            _ => "+OK\r\n".to_string(),
        }
    }
    
    /// 无协调器的集群命令处理
    async fn execute_cluster_command_standalone(&self, cluster_cmd: RedisClusterCommand) -> String {
        match cluster_cmd {
            RedisClusterCommand::Info => {
                "+cluster_state:ok\r\ncluster_slots_assigned:16384\r\ncluster_known_nodes:1\r\n".to_string()
            }
            RedisClusterCommand::Nodes => {
                "+node1 127.0.0.1:6379@16379 master - 0 0 1 connected 0-16383\r\n".to_string()
            }
            RedisClusterCommand::Slots => {
                "*1\r\n*3\r\n:0\r\n:16383\r\n*2\r\n$9\r\n127.0.0.1\r\n:6379\r\n".to_string()
            }
            RedisClusterCommand::MyId => {
                "+node1\r\n".to_string()
            }
            RedisClusterCommand::KeySlot { key } => {
                let slot = self.calculate_slot(&key);
                format!(":{}\r\n", slot)
            }
            _ => "+OK\r\n".to_string(),
        }
    }
    
    /// 计算键的槽位（使用CRC16算法）
    fn calculate_slot(&self, key: &str) -> u16 {
        // 简化的CRC16实现，实际中应该使用正确的Redis CRC16算法
        let mut crc: u16 = 0;
        for byte in key.bytes() {
            crc = crc.wrapping_add(byte as u16);
        }
        crc % 16384
    }
    
    /// 从协调器获取集群槽位信息
    async fn get_cluster_slots_from_coordinator(
        &self,
        coordinator: &Arc<RwLock<ClusterCoordinator>>,
    ) -> String {
        let coord = coordinator.read().await;
        
        // 尝试获取集群的槽位统计信息
        match coord.get_cluster_info().await {
            Ok(cluster_info) => {
                // 如果集群已完全初始化，返回真实的槽位分配
                if cluster_info.slots_assigned == 16384 {
                    // 模拟单节点集群的完整槽位分配
                    let _node_id = coord.get_local_node_id().await;
                    format!(
                        "*1
*3
:0
:16383
*2
$9
127.0.0.1
:6379
"
                    )
                } else {
                    // 槽位未完全分配，返回空数组
                    "*0
".to_string()
                }
            }
            Err(_) => {
                // 无法获取集群信息，返回默认单节点分配
                "*1
*3
:0
:16383
*2
$9
127.0.0.1
:6379
".to_string()
            }
        }
    }
    
    /// 处理普通Redis命令
    async fn handle_redis_command(&self, cmd_name: &str, args: &[String]) -> String {
        // 使用全局存储管理器执行命令
        use crate::storage::GlobalStorageManager;
        
        match GlobalStorageManager::execute_command_async(
            cmd_name.to_string(),
            args.to_vec()
        ).await {
            Ok(result) => {
                // 使用CommandResult自带的to_resp方法
                result.to_resp()
            }
            Err(err) => {
                format!("-ERR {}\r\n", err)
            }
        }
    }
    
    /// 解析RESP命令
    fn parse_resp_command(&self, command_str: &str) -> Vec<String> {
        let lines: Vec<&str> = command_str.split("\r\n").collect();
        let mut args = Vec::new();
        
        if lines.is_empty() || !lines[0].starts_with('*') {
            return args;
        }
        
        // 解析数组长度
        if let Ok(array_len) = lines[0][1..].parse::<usize>() {
            let mut line_idx = 1;
            
            for _ in 0..array_len {
                if line_idx + 1 < lines.len() {
                    // 跳过长度行，取数据行
                    line_idx += 1;
                    if line_idx < lines.len() {
                        args.push(lines[line_idx].to_string());
                    }
                    line_idx += 1;
                }
            }
        }
        
        args
    }
}

/// 集群Redis服务器
pub struct ClusterRedisServer {
    bind_addr: String,
    connection_counter: AtomicU64,
    cluster_coordinator: Option<Arc<RwLock<ClusterCoordinator>>>,
}

impl ClusterRedisServer {
    pub fn new(bind_addr: String) -> Self {
        Self {
            bind_addr,
            connection_counter: AtomicU64::new(0),
            cluster_coordinator: None,
        }
    }
    
    pub fn with_coordinator(mut self, coordinator: Arc<RwLock<ClusterCoordinator>>) -> Self {
        self.cluster_coordinator = Some(coordinator);
        self
    }
    
    pub async fn start(&self) -> io::Result<()> {
        // 初始化全局共享存储
        GlobalStorageManager::initialize();
        
        let listener = TcpListener::bind(&self.bind_addr).await?;
        println!("🚀 集群Redis服务器启动: {}", self.bind_addr);
        
        loop {
            match listener.accept().await {
                Ok((stream, _addr)) => {
                    let connection_id = self.connection_counter.fetch_add(1, Ordering::SeqCst);
                    let coordinator = self.cluster_coordinator.clone();
                    
                    tokio::spawn(async move {
                        if let Err(_e) = Self::handle_connection(connection_id, stream, coordinator).await {
                            // eprintln!("❌ 连接 {} 处理错误: {}", connection_id, e);
                        }
                    });
                }
                Err(_e) => {
                    continue;
                }
            }
        }
    }
    
    async fn handle_connection(
        connection_id: ConnectionId, 
        stream: TcpStream,
        coordinator: Option<Arc<RwLock<ClusterCoordinator>>>,
    ) -> io::Result<()> {
        let network_stream = ClusterTokioStream::from_tokio_stream(stream)?;
        
        let mut protocol_handler = ClusterProtocolHandler::new(connection_id);
        if let Some(coord) = coordinator {
            protocol_handler = protocol_handler.with_coordinator(coord);
        }
        
        protocol_handler.handle_connection_stream(Box::new(network_stream)).await
    }
}

/// 启动集群Redis服务器
pub async fn start_cluster_redis_server(
    bind_addr: &str,
    coordinator: Option<Arc<RwLock<ClusterCoordinator>>>,
) -> io::Result<()> {
    let mut server = ClusterRedisServer::new(bind_addr.to_string());
    if let Some(coord) = coordinator {
        server = server.with_coordinator(coord);
    }
    server.start().await
}